package com.dtwave.lnstreaming.process.function.process;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class XfaceKeyProcessFunction extends KeyedProcessFunction<String, JSONObject,JSONObject> {


    private final MapStateDescriptor<String, JSONObject> userDataDescriptor = new MapStateDescriptor<>(
            "user-state",
            Types.STRING,
            TypeInformation.of(JSONObject.class)
    );
    private MapState<String,JSONObject> userState;
    public long sinkTime;

    //开启
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        long ttlTime = 30;
        sinkTime = 2000L;
        StateTtlConfig ttlConfig = StateTtlConfig
                .newBuilder(Time.minutes(ttlTime)).setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                .build();
        userDataDescriptor.enableTimeToLive(ttlConfig);
        userState = getRuntimeContext().getMapState(userDataDescriptor);
    }

    //关闭
    @Override
    public void close() throws Exception {
        super.close();
    }

    //延时判断处理
    @Override
    public void processElement(JSONObject jsonObject, Context context, Collector<JSONObject> collector) throws Exception {
        //不加任何判断让所有数据延迟2s发送
        userState.put(context.getCurrentKey(),jsonObject);
        context.timerService().registerProcessingTimeTimer(System.currentTimeMillis()+sinkTime);
    }

    //定时器到期
    @Override
    public void onTimer(long timestamp, KeyedProcessFunction<String, JSONObject, JSONObject>.OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
        //删除定时器
        ctx.timerService().deleteProcessingTimeTimer(timestamp);
        //返回resultdata
        JSONObject jsonObject = userState.get(ctx.getCurrentKey());
        if(jsonObject!=null){
            out.collect(jsonObject);
        }
    }
}
